package org.example.branch_tag;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class CreateBranch {
    public static void main(String[] args) throws Exception {
        // 配置Flink Web UI端口
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT, "8081");
        
        // 创建Batch Table环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        // 创建Paimon catalog
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println("正在执行SQL: \n" + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        // 查询所有分支
        String queryBranchesSQL = "SELECT * FROM `paimon`.`default`.`LogTable$branches`;";
        System.out.println("正在执行SQL: \n" + queryBranchesSQL);
        tableEnv.executeSql(queryBranchesSQL).print();

        // 查询默认分支的数据
        String queryDefaultBranchDataSQL = "SELECT count(1) FROM `paimon`.`default`.`LogTable`;";
        System.out.println("正在执行SQL: \n" + queryDefaultBranchDataSQL);
        tableEnv.executeSql(queryDefaultBranchDataSQL).print();

        // 将main分支数据插入到log_branch分支
        String insertBranchDataSQL = "INSERT INTO `paimon`.`default`.`LogTable$branch_log_branch`\n" +
                "SELECT * FROM `paimon`.`default`.`LogTable`;";
        System.out.println("正在执行SQL: \n" + insertBranchDataSQL);
        tableEnv.executeSql(insertBranchDataSQL);

        // 查询log_branch分支的数据
        String queryBranchDataSQL = "SELECT count(1) FROM `paimon`.`default`.`LogTable$branch_log_branch`;";
        System.out.println("正在执行SQL: \n" + queryBranchDataSQL);
        tableEnv.executeSql(queryBranchDataSQL).print();
    }
}
